/*
    asyframe for asyframe@gmail.com

    Copyright (C) 2011, asyframe@gmail.com, http://asyframe.googlecode.com/

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
	(at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
	along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "stdafx.h"
#include "ConstDef.h"
#include "Common/Logger.h"
#include "Common/Interface.h"
#include "Common/ThreadHandler.h"
#include "Common/WorkerBase.h"
#include "Common/AppBase.h"
#include "Common/BufferPool.h"
#include "Common/WorkerBase.h"

namespace asyframe
{
	WorkerBase::WorkerBase(std::size_t _indexInPool)
		:indexInPool_(_indexInPool),pIoService_(),pBufferPool_(),pWork_(),pThread_()
	{

	}

	WorkerBase::~WorkerBase()
	{
		pIoService_.reset();
		pBufferPool_.reset();
		pWork_.reset();
		pThread_.reset();
	}

	boost::shared_ptr<boost::asio::io_service> WorkerBase::getIoService()
	{
		return pIoService_;
	}

	boost::shared_ptr<BufferPool> WorkerBase::getBufferPool()
	{
		return pBufferPool_;
	}

	boost::shared_ptr<boost::thread> WorkerBase::getThread()
	{
		return pThread_;
	}

	boost::shared_ptr<BufferPool> WorkerBase::makeBufferPool()
	{
		boost::shared_ptr<BufferPool> pBufferPool(new BufferPool(shared_from_this()));
		return pBufferPool;
	}

	std::size_t WorkerBase::getID()
	{
		return indexInPool_;
	}

	bool WorkerBase::isInWorkerThread()
	{
		return pThread_ && pThread_->get_id() == boost::this_thread::get_id();
	}

	retT WorkerBase::init()
	{
		pIoService_.reset(new boost::asio::io_service);
		pBufferPool_ = makeBufferPool();
		RETSUCCESS;
	}

	WorkerPoolBase::WorkerPoolBase( std::size_t _poolSize )
		:workerPool_(),nextWorker_(0),block_(false),started_(false)
	{
		BOOST_ASSERT(_poolSize != 0);

		for (std::size_t i = 0; i < _poolSize; ++i)
		{
			workerPool_.push_back(makeWorker(i));
		}
	}

	WorkerPoolBase::~WorkerPoolBase()
	{
		for (std::size_t i = 0; i < workerPool_.size(); ++i)
		{
			if(workerPool_[i])
			{
				workerPool_[i]->pWork_.reset();
				if(workerPool_[i]->pIoService_)
				{
					workerPool_[i]->pIoService_->stop();
					workerPool_[i]->pIoService_.reset();
				}
				if(workerPool_[i]->pThread_)
				{
					workerPool_[i]->pThread_->interrupt();
					workerPool_[i]->pThread_.reset();
				}
			}
		}
		workerPool_.clear();

		nextWorker_ = 0;
		block_ = false;
		started_=false;
	}

	std::size_t WorkerPoolBase::size()
	{
		return workerPool_.size();
	}

	void WorkerPoolBase::start()
	{
		start(false);
	}

	void WorkerPoolBase::run()
	{
		start(true);
	}

	void WorkerPoolBase::stop()
	{
		if (!started_)
			return;

		for (std::size_t i = 0; i < workerPool_.size(); ++i)
		{
			if(workerPool_[i])
				workerPool_[i]->pWork_.reset();
		}

		if (!block_)
			wait();
		started_ = false;
	}

	void WorkerPoolBase::forceStop()
	{
		for (std::size_t i = 0; i < workerPool_.size(); ++i)
		{
			if(workerPool_[i])
			{
				workerPool_[i]->pWork_.reset();
				if(workerPool_[i]->pIoService_)
					workerPool_[i]->pIoService_->stop();
				if(workerPool_[i]->pThread_)
					workerPool_[i]->pThread_->interrupt();
			}
		}
		started_=false;
	}

	boost::shared_ptr<WorkerBase> WorkerPoolBase::getWorker()
	{
		boost::shared_ptr<WorkerBase> pWorker = workerPool_[nextWorker_];
		if (++nextWorker_ == workerPool_.size())
			nextWorker_ = 0;
		return pWorker;
	}

	boost::shared_ptr<WorkerBase> WorkerPoolBase::getWorker( std::size_t _offset )
	{
		if (_offset < workerPool_.size())
			return workerPool_[_offset];
		else
			return getWorker();
	}

	void WorkerPoolBase::wait()
	{
		if (!started_)
			return;

		for (std::size_t i = 0; i < workerPool_.size(); ++i)
		{
			if(workerPool_[i]->pThread_)
			{
				workerPool_[i]->pThread_->join();
				workerPool_[i]->pThread_.reset();
			}
		}
	}

	void WorkerPoolBase::start( bool _block )
	{
		if (started_)
			return;
		started_ = true;

		for (std::size_t i = 0; i < workerPool_.size(); ++i)
		{
			if(workerPool_[i] && workerPool_[i]->pIoService_)
			{
				workerPool_[i]->pIoService_->reset();
				workerPool_[i]->pWork_.reset(new boost::asio::io_service::work(*workerPool_[i]->pIoService_));
				workerPool_[i]->pThread_.reset(new boost::thread(
					boost::bind(&boost::asio::io_service::run, workerPool_[i]->pIoService_)));
			}
		}

		block_ = _block;

		if (_block)
			wait();
	}

	boost::shared_ptr<WorkerBase> WorkerPoolBase::makeWorker(std::size_t _index)
	{
		boost::shared_ptr<WorkerBase> pWorker(new WorkerBase(_index));
		return pWorker;
	}

	retT WorkerPoolBase::init()
	{
		for (std::size_t i = 0; i < workerPool_.size(); ++i)
		{
			if(workerPool_[i])
			{
				R(workerPool_[i]->init());
			}
		}
		RETSUCCESS;
	}

	std::vector< boost::shared_ptr<WorkerBase> >& WorkerPoolBase::getWorkerPool()
	{
		return workerPool_;
	}

} /* namespace asyframe */
